import datetime
from urllib import parse

import pandas as pd
import pymysql
from dbutils.pooled_db import PooledDB
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


# sql执行过程中，不允许引发任何其他非该类型异常，在函数执行finally中关闭，捕获异常rollback
class SqlError(Exception):
    pass


class SqlAlchemyConn(object):
    """
    get alchemy connection instance
    eg: conn = SqlAlchemyConn(**kwargs)
    """

    sql_alchemy_conn = None
    sql_session = None
    mydb_con = None

    def __new__(cls, *args, **kwargs):
        # if not cls.sql_alchemy_conn:
        cls.sql_alchemy_conn = SqlAlchemyConn.__get_alchemy_conn(**kwargs)
        return cls.sql_alchemy_conn

    @staticmethod
    def __get_alchemy_conn(**kwargs):
        """
        get mysql connection instance
        :param kwargs:
        :return:
        """
        return create_engine(
            "mysql+pymysql://{}:{}@{}:{}/{}?charset={}".format(
                kwargs.get("MYSQL_USER"),
                parse.quote_plus(kwargs.get("MYSQL_PASSWORD")),
                kwargs.get("MYSQL_HOST"),
                int(kwargs.get("MYSQL_PORT")),
                kwargs.get("MYSQL_DB_NAME"),
                kwargs.get("MYSQL_ENCODING"),
            ),
            pool_size=20,
            pool_recycle=3600,
            max_overflow=0,
            isolation_level="AUTOCOMMIT",
        )

    @classmethod
    def get_alchemy_session(cls, engine):
        print("get sql session")
        if not cls.sql_session:
            cls.sql_session = sessionmaker(bind=engine)()
        return cls.sql_session


class MysqlConn(object):
    """
    get mysql connection instance
    """

    mysql_conn = None
    mydb_conn = None

    def __new__(cls, *args, **kwargs):
        if not cls.mysql_conn:
            cls.mysql_conn = MysqlConn.__get_pymysql_con(**kwargs)
        return cls.mysql_conn

    @staticmethod
    def __get_pymysql_con(**kwargs):
        """
        获取MySQL引擎
        :param kwargs:
        :return:
        """
        return pymysql.connect(
            host=kwargs.get("MYSQL_HOST"),
            user=kwargs.get("MYSQL_USER"),
            password=kwargs.get("MYSQL_PASSWORD"),
            port=int(kwargs.get("MYSQL_PORT")),
            database=kwargs.get("MYSQL_DB_NAME"),
            use_unicode=True,
            charset="utf8",
        )

    @classmethod
    def get_mydb_con(cls, kwargs):
        if not cls.mydb_conn:
            cls.mydb_conn = pymysql.connect(
                host=kwargs.get("MYSQL_HOST"),
                user=kwargs.get("MYSQL_USER"),
                password=kwargs.get("MYSQL_PASSWORD"),
                port=int(kwargs.get("MYSQL_PORT")),
                db=kwargs.get("MYSQL_DB_NAME"),
                charset="utf8",
            )
        return cls.mydb_conn


# class MysqlPool:
#     def __new__(cls, *args, **kwargs):
#         if not hasattr(cls, 'instance'):
#             cls.instance = super(MysqlPool, cls).__new__(cls)
#         return cls.instance
#
#     def __init__(self, host, port, user, password, database, charset='utf8'):
#         self._pool = PooledDB(creator=pymysql, maxcached=20, mincached=1, host=host, port=port, user=user,
#                               password=password, database=database, charset=charset,
#                               cursorclass=pymysql.cursors.DictCursor)
#
#     def get_connection(self):
#         return self._pool.connection()
#
#     def get(self, sql, params=None):
#         con = self.get_connection()
#         cursor = con.cursor()
#         try:
#             if params:
#                 cursor.execute(sql, params)
#             else:
#                 cursor.execute(sql)
#
#             result = cursor.fetchall()
#             if result == ():
#                 result = []
#             for i in result:
#                 for _k, _v in i.items():
#                     if isinstance(_v, datetime.datetime) or isinstance(_v, datetime.date):
#                         i[_k] = str(_v)
#             return result
#         finally:
#             cursor.close()
#             con.close()
#
#     def execute(self, sql_list: list):
#         con = self._pool.connection()
#         cursor = con.cursor()
#         try:
#             for i in sql_list:
#                 sql = i['sql']
#                 value = i['value']
#                 if not isinstance(value, list):
#                     raise TypeError(f"The type of value is not list: {value}")
#                 cursor.execute(sql, value)
#             con.commit()
#             return True
#         except Exception as e:
#             con.rollback()
#             # 只有这样报错才能返回，不会无限运行
#             raise SqlError(e)
#         finally:
#             cursor.close()
#             con.close()
#
#     def get_cursor(self):
#         # 针对一些复杂的既需要查询又需要插入、更新的事务，返回cursor，自己进行管理
#         con = self._pool.connection()
#         cursor = con.cursor()
#         return con, cursor
#
#     def execute_sql_curosr(self, cursor, sql, value=None):
#         # 与get_cursor配合使用，其实也不至于，都是一条语句，难以简化
#         try:
#             if value is None:
#                 cursor.execute(sql)
#             else:
#                 cursor.execute(sql, value)
#             return True
#         except Exception as e:
#             # 只有这样报错才能返回，不会无限运行
#             raise SqlError(e)
#
#     def get_df(self, sql, params=None):
#         con = self.get_connection()
#         try:
#             result = pd.read_sql(sql, con, params)
#             return result
#         finally:
#             con.close()


class MysqlPool:
    def __init__(self, host, port, user, password, database, charset="utf8"):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database
        self.charset = charset

    def get(self, sql, params=None):
        con = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,  # 在这里输入用户名
            password=self.password,  # 在这里输入密码
            database=self.database,
            charset=self.charset,
            cursorclass=pymysql.cursors.DictCursor,
        )
        cursor = con.cursor()
        try:
            if params:
                cursor.execute(sql, params)
            else:
                cursor.execute(sql)

            result = cursor.fetchall()
            if result == ():
                result = []
            for i in result:
                for _k, _v in i.items():
                    if isinstance(_v, datetime.datetime) or isinstance(
                        _v, datetime.date
                    ):
                        i[_k] = str(_v)
            return result
        finally:
            cursor.close()
            con.close()

    def execute(self, sql_list: list):
        con = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,  # 在这里输入用户名
            password=self.password,  # 在这里输入密码
            database=self.database,
            charset=self.charset,
            cursorclass=pymysql.cursors.DictCursor,
        )
        cursor = con.cursor()
        result = []
        try:
            for i in sql_list:
                sql = i["sql"]
                value = i["value"]
                if not isinstance(value, list):
                    raise TypeError(f"The type of value is not list: {value}")
                tmp_res = cursor.execute(sql, value)
                result.append(tmp_res)
            con.commit()
        except Exception as e:
            con.rollback()
            # 只有这样报错才能返回，不会无限运行
            raise SqlError(e)
        finally:
            cursor.close()
            con.close()
            return result

    def get_cursor(self):
        # 针对一些复杂的既需要查询又需要插入、更新的事务，返回cursor，自己进行管理
        con = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,  # 在这里输入用户名
            password=self.password,  # 在这里输入密码
            database=self.database,
            charset=self.charset,
            cursorclass=pymysql.cursors.DictCursor,
        )
        cursor = con.cursor()
        return con, cursor

    def execute_sql_curosr(self, cursor, sql, value=None):
        # 与get_cursor配合使用，其实也不至于，都是一条语句，难以简化
        try:
            if value is None:
                cursor.execute(sql)
            else:
                cursor.execute(sql, value)
            return True
        except Exception as e:
            # 只有这样报错才能返回，不会无限运行
            raise SqlError(e)

    def get_df(self, sql, params=None):
        con = pymysql.connect(
            host=self.host,
            port=self.port,
            user=self.user,  # 在这里输入用户名
            password=self.password,  # 在这里输入密码
            database=self.database,
            charset=self.charset,
            cursorclass=pymysql.cursors.DictCursor,
        )
        try:
            result = pd.read_sql(sql, con, params)
            return result
        finally:
            con.close()
